package org.example.client;

import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.data.Stat;
import org.example.ServerConnection;
import org.example.common.ZKConnectionHelper;
import org.example.protocol.RpcRequest;
import org.example.protocol.RpcResponse;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * rpc的调用者
 */
public class RpcInvoker {
    // 1、创建一个线程池对象，处理自定义事件
    private final ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    // 定义一个Map，保存<ip:port,连接>的键值对
    private Map<String, ServerConnection> serverMap;

    private ZkClient zk;

    private final String clientZKPath = ZKConnectionHelper.clientPath + "/" + UUID.randomUUID().toString();

    public RpcInvoker() throws InterruptedException {
        initServerConnection();
        startMonitorThread();
    }

    // 监控线程
    public void startMonitorThread() throws InterruptedException {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    for (String child : zk.getChildren(clientZKPath)) {
                        String childPath = clientZKPath + "/" + child;
                        Stat stat = new Stat();
                        zk.readData(childPath, stat);
                        if (System.currentTimeMillis() - stat.getMtime() > 5000) {
                            zk.delete(childPath);
                        }
                    }
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();
    }

    // 3、初始化客户端(创建连接池，创建bootstrap，设置bootstrap，连接服务器)
    public void initServerConnection() throws InterruptedException {
        serverMap = new ConcurrentHashMap<>();

        // 初始化ZK客户端
        zk = new ZkClient(ZKConnectionHelper.zkUrl);

        System.out.println("Client ZK Path: " + clientZKPath);

        // 为客户端创建节点
        zk.createPersistent(clientZKPath, true);

        // 查询zk的/rpc/server下有多少个server，为每一个server创建连接
        List<String> children = zk.getChildren(ZKConnectionHelper.serverPath);

        // 遍历children列表，进行ServerConnection的创建
        for (String child : children) {
            System.out.println("new server: " + child + " found!");
            ServerConnection serverConnection = new ServerConnection(child);
            serverMap.put(child, serverConnection);
        }

        // 注册server节点的子节点监听
        zk.subscribeChildChanges(ZKConnectionHelper.serverPath, new IZkChildListener() {
            @Override
            public void handleChildChange(String parentPath, List<String> currentChildren) throws Exception {
                Set<String> oldChildren = serverMap.keySet();
                for (String child : currentChildren) {
                    if (!oldChildren.contains(child)) {
                        System.out.println("new server: " + child + " found!");
                        ServerConnection serverConnection = new ServerConnection(child);
                        serverMap.put(child, serverConnection);
                    }
                }

                for (String oldChild : oldChildren) {
                    if (!currentChildren.contains(oldChild)) {
                        System.out.println("lost server: " + oldChild + "!");
                        serverMap.remove(oldChild);
                    }
                }

            }
        });
    }

    // 4、编写一个方法，使用JDK的动态代理创建对象
    public Object createProxy(final Class<?> serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{serviceClass}, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                // 2) rpcResponseHandler设置请求报文参数
                RpcRequest rpcRequest = new RpcRequest();
                rpcRequest.setRequestId("1");
                rpcRequest.setClassName(serviceClass.getName());
                rpcRequest.setMethodName(method.getName());
                rpcRequest.setParameterTypes(method.getParameterTypes());
                rpcRequest.setParameters(args);

                // 这里改成从zk中获取各服务端的响应时间
                // 判断响应时间来进行请求，逻辑是
                // 优先选择没有记录响应时间的服务器
                // 如果都有响应时间，则选择响应时间短的那一个
                ServerConnection selectServerConnection = null;
                long minCostTime = System.currentTimeMillis();
                for (ServerConnection serverConnection : serverMap.values()) {
                    String path = clientZKPath + "/" + serverConnection.getIndex();
                    // 如果这个服务器没有记录时间，直接选中这个
                    if (!zk.exists(path)) {
                        selectServerConnection = serverConnection;
                        break;
                    }

                    // 找响应时间最短的服务器
                    Long costTime = zk.readData(path);
                    if (costTime < minCostTime) {
                        minCostTime = costTime;
                        selectServerConnection = serverConnection;
                    }
                }

                if (selectServerConnection != null) {
                    System.out.println("Select Server: " + selectServerConnection.getIndex());
                    selectServerConnection.getRpcResponseHandler().setRpcRequest(rpcRequest);
                    long startTime = System.currentTimeMillis();
                    RpcResponse result = (RpcResponse) executorService.submit(selectServerConnection.getRpcResponseHandler()).get(300, TimeUnit.MILLISECONDS);

                    // 结果返回以后，要将服务器的响应时间记录到zk中，以供后面的请求参考
                    // 先要判断节点是否存在
                    String path = clientZKPath + "/" + selectServerConnection.getIndex();
                    if (!zk.exists(path)) {
                        zk.createPersistent(path, System.currentTimeMillis() - startTime);
                    } else {
                        zk.writeData(path, System.currentTimeMillis() - startTime);
                    }

                    // 4）返回结果
                    return result.getResult();
                }
                return "No Server Can Request";
            }
        });
    }

}
